package com.heima.article.listeners;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.article.service.ApArticleConfigService;
import com.heima.article.service.ApArticleService;
import com.heima.article.service.ApAuthorService;
import com.heima.article.service.HotArticleService;
import com.heima.common.constant.GlobalConstant;
import com.heima.model.article.dots.ReadBehaviorDto;
import com.heima.model.article.dots.UpdateArticleMsg;
import com.heima.model.article.pojos.ApArticle;
import com.heima.model.article.pojos.ApAuthor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
@Slf4j
public class KafkaListeners {
    @Autowired
    ApArticleConfigService apArticleConfigService;
    @Autowired
    ApArticleService apArticleService;
    @Autowired
    ApAuthorService apAuthorService;
    @Autowired
    HotArticleService hotArticleService;
    @KafkaListener(topics= GlobalConstant.UP_OR_DOWN_TOPIC)
    public void upOrDownListener(String msg)
    {
        if (StringUtils.isNotBlank(msg))
        {
            log.info("接收消息->"+msg);
            Map map = JSON.parseObject(msg, Map.class);
            apArticleConfigService.updateEnableByMap(map);
        }
    }
    @KafkaListener(topics = GlobalConstant.USER_LIKE)
    public void userLike(String msg)
    {
        if (StringUtils.isNotBlank(msg))
        {
            log.info("主题 {}：收到消息：{}",GlobalConstant.USER_LIKE,msg);
            Long articleId=Long.valueOf(msg);
            apArticleService.update(Wrappers.<ApArticle>lambdaUpdate().eq(ApArticle::getId,articleId).setSql("likes=likes+1"));
        }
    }
    @KafkaListener(topics = GlobalConstant.USER_CANCEL_LIKE)
    public void userCancelLike(String msg)
    {
        if (StringUtils.isNotBlank(msg))
        {
            log.info("主题 {}：收到消息：{}",GlobalConstant.USER_CANCEL_LIKE,msg);
            Long articleId=Long.valueOf(msg);
            apArticleService.update(Wrappers.<ApArticle>lambdaUpdate().eq(ApArticle::getId,articleId).setSql("likes=likes-1"));
        }
    }
//    @KafkaListener(topics = GlobalConstant.USER_COLLECTION)
//    public void userCollection(String msg)
//    {
//        if (StringUtils.isNotBlank(msg))
//        {
//            log.info("主题 {}：收到消息：{}",GlobalConstant.USER_COLLECTION,msg);
//            Long articleId=Long.valueOf(msg);
//            apArticleService.update(Wrappers.<ApArticle>lambdaUpdate().eq(ApArticle::getId,articleId).setSql("collection=collection+1"));
//        }
//    }
//    @KafkaListener(topics = GlobalConstant.USER_CANCEL_COLLECTION)
//    public void userCancelCollection(String msg)
//    {
//        if (StringUtils.isNotBlank(msg))
//        {
//            log.info("主题 {}：收到消息：{}",GlobalConstant.USER_CANCEL_COLLECTION,msg);
//            Long articleId=Long.valueOf(msg);
//            apArticleService.update(Wrappers.<ApArticle>lambdaUpdate().eq(ApArticle::getId,articleId).setSql("collection=collection-1"));
//        }
//    }
    @KafkaListener(topics = GlobalConstant.USER_READ)
    public void userRead(String msg)
    {
        if (StringUtils.isNotBlank(msg))
        {
            log.info("主题 {}：收到消息：{}",GlobalConstant.USER_READ,msg);
            ReadBehaviorDto dto = JSON.parseObject(msg, ReadBehaviorDto.class);
            apArticleService.update(Wrappers.<ApArticle>lambdaUpdate().eq(ApArticle::getId,dto.getArticleId()).setSql("views=views+"+dto.getCount()));
        }
    }
    @KafkaListener(topics = GlobalConstant.ADD_AP_AUTHOR)
    public void addApAuthor(String msg)
    {
        if (StringUtils.isNotBlank(msg))
        {
            log.info("主题 {}：收到消息：{}",GlobalConstant.ADD_AP_AUTHOR,msg);
            ApAuthor apAuthor = JSON.parseObject(msg, ApAuthor.class);
            apAuthorService.save(apAuthor);
        }
    }
    @KafkaListener(topics = GlobalConstant.KAFKA_STREAM_BATCH_TOPIC)
    public void batch(String msg)
    {
        if(StringUtils.isNotBlank(msg))
        {
            log.info("主题 {}：收到消息：{}",GlobalConstant.KAFKA_STREAM_BATCH_TOPIC,msg);
            UpdateArticleMsg updateArticleMsg = JSON.parseObject(msg, UpdateArticleMsg.class);
            hotArticleService.updateHotArticle(updateArticleMsg);
        }
    }


}
